home *** CD-ROM | disk | FTP | other *** search
- # Source Generated with Decompyle++
- # File: in.pyc (Python 2.6)
-
- __author__ = 'api.jscudder (Jeff Scudder)'
- import atom.http_interface as atom
- import atom.url as atom
-
- class Error(Exception):
- pass
-
-
- class NoRecordingFound(Error):
- pass
-
-
- class MockRequest(object):
- '''Holds parameters of an HTTP request for matching against future requests.
- '''
-
- def __init__(self, operation, url, data = None, headers = None):
- self.operation = operation
- if isinstance(url, (str, unicode)):
- url = atom.url.parse_url(url)
-
- self.url = url
- self.data = data
- self.headers = headers
-
-
-
- class MockResponse(atom.http_interface.HttpResponse):
- '''Simulates an httplib.HTTPResponse object.'''
-
- def __init__(self, body = None, status = None, reason = None, headers = None):
- if body and hasattr(body, 'read'):
- self.body = body.read()
- else:
- self.body = body
- if status is not None:
- self.status = int(status)
- else:
- self.status = None
- self.reason = reason
- if not headers:
- pass
- self._headers = { }
-
-
- def read(self):
- return self.body
-
-
-
- class MockHttpClient(atom.http_interface.GenericHttpClient):
-
- def __init__(self, headers = None, recordings = None, real_client = None):
- """An HttpClient which responds to request with stored data.
-
- The request-response pairs are stored as tuples in a member list named
- recordings.
-
- The MockHttpClient can be switched from replay mode to record mode by
- setting the real_client member to an instance of an HttpClient which will
- make real HTTP requests and store the server's response in list of
- recordings.
-
- Args:
- headers: dict containing HTTP headers which should be included in all
- HTTP requests.
- recordings: The initial recordings to be used for responses. This list
- contains tuples in the form: (MockRequest, MockResponse)
- real_client: An HttpClient which will make a real HTTP request. The
- response will be converted into a MockResponse and stored in
- recordings.
- """
- if not recordings:
- pass
- self.recordings = []
- self.real_client = real_client
- if not headers:
- pass
- self.headers = { }
-
-
- def add_response(self, response, operation, url, data = None, headers = None):
- '''Adds a request-response pair to the recordings list.
-
- After the recording is added, future matching requests will receive the
- response.
-
- Args:
- response: MockResponse
- operation: str
- url: str
- data: str, Currently the data is ignored when looking for matching
- requests.
- headers: dict of strings: Currently the headers are ignored when
- looking for matching requests.
- '''
- request = MockRequest(operation, url, data = data, headers = headers)
- self.recordings.append((request, response))
-
-
- def request(self, operation, url, data = None, headers = None):
- """Returns a matching MockResponse from the recordings.
-
- If the real_client is set, the request will be passed along and the
- server's response will be added to the recordings and also returned.
-
- If there is no match, a NoRecordingFound error will be raised.
- """
- if self.real_client is None:
- if isinstance(url, (str, unicode)):
- url = atom.url.parse_url(url)
-
- for recording in self.recordings:
- if recording[0].operation == operation and recording[0].url == url:
- return recording[1]
-
- raise NoRecordingFound('No recodings found for %s %s' % (operation, url))
- self.real_client is None
- response = self.real_client.request(operation, url, data = data, headers = headers)
- stored_response = MockResponse(body = response, status = response.status, reason = response.reason)
- self.add_response(stored_response, operation, url, data = data, headers = headers)
- return stored_response
-
-
-